"""
Case Type   : 基础功能
Case Name   : 同步基础数据时存在mot表
Create At   : 2022/11/08
Owner       : opentestcase026
Description :
    1.两个集群均创建表（mot表与astore表）
    2.创建发布端
    3.创建订阅端
    4.发布端插入基础数据
    5.查询订阅端表数据
      5.1 查询mot表表样数据
      5.2 查询astore表表样数据
    6.删除订阅
    7.删除发布
    8.清理环境
Expect      :
    1.成功
    2.成功
    3.成功
    4.成功
    5.数据同步成功
      5.1 mot表表样数据未同步成功
      5.2 astore表样数据同步成功
    6.成功
    7.成功
    8.成功
History     :
"""
import unittest
import os
from yat.test import macro
from yat.test import Node
from testcase.utils.Logger import Logger
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Common import Common
from testcase.utils.Constant import Constant

Primary_SH = CommonSH('PrimaryDbUser')


@unittest.skipIf(3 != Primary_SH.get_node_num(), '非1+2环境不执行')
class Pubsubclass(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info("-----this is setup-----")
        self.log.info(f"-----{os.path.basename(__file__)[:-3]} start-----")
        self.pri_userdb_pub = Node(node='PrimaryDbUser')
        self.pri_userdb_sub = Node(node='remote1_PrimaryDbUser')
        self.constant = Constant()
        self.commsh_pub = CommonSH('PrimaryDbUser')
        self.commsh_sub = CommonSH('remote1_PrimaryDbUser')
        self.com_pub = Common()
        self.com_sub = Common()
        self.tb_name1 = 'tb_pubsub_case234_01'
        self.tb_name2 = 'tb_pubsub_case234_02'
        self.subname = "sub_case234"
        self.pubname = "pub_case234"
        self.port = str(int(self.pri_userdb_pub.db_port) + 1)
        self.wal_level = self.com_pub.show_param("wal_level")
        self.pub_incremental_param = \
            self.com_pub.show_param("enable_incremental_checkpoint")
        self.log.info(self.pub_incremental_param)
        self.sub_incremental_param = \
            self.com_sub.show_param("enable_incremental_checkpoint")
        self.log.info(self.sub_incremental_param)
        self.user_param_pub = f'-U {self.pri_userdb_pub.db_user} ' \
                              f'-W {self.pri_userdb_pub.db_password}'
        self.user_param_sub = f'-U {self.pri_userdb_sub.db_user} ' \
                              f'-W {self.pri_userdb_sub.db_password}'
        text = '-----修改pg_hba expect:成功-----'
        self.log.info(text)
        self.log.info('-----发布端状态-----')
        status = self.commsh_pub.get_db_cluster_status('detail')
        self.log.info(status)
        self.assertEqual(status.count('Normal'), 4, '执行失败' + text)
        self.log.info('-----订阅端状态-----')
        status = \
            self.commsh_sub.get_db_cluster_status(
                'detail', env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(status)
        self.assertEqual(status.count('Normal'), 4, '执行失败' + text)
        guc_res = self.commsh_pub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, '',
            f'host    replication  {self.pri_userdb_sub.db_user} '
            f'{self.pri_userdb_sub.db_host}/32 sha256')
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败:' + text)
        if 'logical' != self.wal_level:
            result = self.commsh_pub.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG, 'wal_level=logical')
            self.assertTrue(result, '执行失败:' + text)
            result = self.commsh_pub.restart_db_cluster(True)
            flg = self.constant.START_SUCCESS_MSG in result \
                  or 'Degrade' in result
            self.assertTrue(flg, '执行失败:' + text)
        if 'off' != self.pub_incremental_param:
            result = self.commsh_pub.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG,
                'enable_incremental_checkpoint=off')
            self.assertTrue(result, '执行失败:' + text)
            restart = self.commsh_pub.restart_db_cluster(True)
            flg = self.constant.START_SUCCESS_MSG in restart \
                  or 'Degrade' in restart
            self.assertTrue(flg, '执行失败:' + text)
        if 'off' != self.sub_incremental_param:
            result = self.pri_userdb_sub.sh(
                f'''source {macro.DB_ENV_PATH_REMOTE1}; \
                gs_guc set -N all -D {macro.DB_INSTANCE_PATH_REMOTE1} \
                -c "enable_incremental_checkpoint=off"''').result()
            self.assertIn('ALL: Success to perform gs_guc', result,
                          '订阅端MOT参数设置失败')
            restart = self.pri_userdb_sub.sh(
                f'''source {macro.DB_ENV_PATH_REMOTE1};
                gs_om -t restart''').result()
            flg = self.constant.START_SUCCESS_MSG in restart \
                  or 'Degrade' in restart
            self.assertTrue(flg, '执行失败:' + text)
        guc_res = self.commsh_sub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH_REMOTE1,
            f'host    replication  {self.pri_userdb_pub.db_user} '
            f'{self.pri_userdb_pub.db_host}/32 sha256',
            macro.DB_ENV_PATH_REMOTE1)
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败:' + text)

    def test_pubsub(self):
        text = '-----step1:两个集群均创建表 expect:成功-----'
        self.log.info(text)
        sql = f'''create foreign table {self.tb_name1}
                  (i int primary key, t varchar(10));
                  create table {self.tb_name2}
                  (i int primary key, t text) 
                  WITH (STORAGE_TYPE='astore');'''
        pub_result = self.commsh_pub.execut_db_sql(sql, self.user_param_pub)
        self.log.info(pub_result)
        self.assertEqual(pub_result.count(self.constant.TABLE_CREATE_SUCCESS),
                         2, '执行失败:' + text)
        self.assertEqual(pub_result.count(
            self.constant.CREATE_FOREIGN_SUCCESS_MSG), 2, '执行失败:' + text)
        sub_result = self.commsh_sub.execut_db_sql(
            sql, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE1)
        self.log.info(sub_result)
        self.assertEqual(sub_result.count(self.constant.TABLE_CREATE_SUCCESS),
                         2, '执行失败:' + text)
        self.assertEqual(sub_result.count(
            self.constant.CREATE_FOREIGN_SUCCESS_MSG), 2, '执行失败:' + text)

        text = '-----step2:创建发布端;expect:成功-----'
        self.log.info(text)
        sql = f"CREATE PUBLICATION {self.pubname} " \
              f"FOR all TABLEs;"
        result = self.commsh_pub.execut_db_sql(sql,
                                               sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertIn(self.constant.create_pub_succ_msg, result,
                      '执行失败:' + text)
        self.assertNotIn(self.constant.SQL_WRONG_MSG[1], result,
                         '执行失败:' + text)

        text = '-----step3:创建订阅端;expect:成功-----'
        self.log.info(text)
        result = self.commsh_sub.execute_generate(
            macro.COMMON_PASSWD, env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertIn('', result, '执行失败:' + text)
        sql = f"CREATE SUBSCRIPTION {self.subname} CONNECTION " \
              f"'host={self.pri_userdb_pub.db_host} " \
              f"port={self.port} " \
              f"user={self.pri_userdb_pub.db_user} " \
              f"dbname={self.pri_userdb_pub.db_name} " \
              f"password={self.pri_userdb_pub.ssh_password}' " \
              f"PUBLICATION {self.pubname} with (copy_data=false);"
        result = self.commsh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertIn(self.constant.create_sub_succ_msg,
                      result, '执行失败:' + text)

        text = '-----step4:发布端插入数据 expect:成功-----'
        self.log.info(text)
        sql = f'''insert into {self.tb_name1} values(1, 'first');
                 insert into {self.tb_name2} values(1, 'first');'''
        result = self.commsh_pub.execut_db_sql(sql,
                                               sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertIn(self.constant.INSERT_SUCCESS_MSG, result, '执行失败' +
                      text)

        text = '-----step5:查询订阅端表数据 expect:成功-----'
        self.log.info(text)
        text = '-----step5.1:查询订阅端mot表数据 expect:mot表数据未同步成功-----'
        self.log.info(text)
        sql = f"select * from {self.tb_name1};"
        result = self.commsh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertEqual(result.count('0 row'), 1, '执行失败' + text)

        text = '-----step5.2:查询订阅端astore表数据 expect:astore表数据同步成功-----'
        self.log.info(text)
        sql = f"select * from {self.tb_name2};"
        result = self.commsh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertEqual(result.count('1 row'), 1, '执行失败' + text)
        self.assertIn(' 1 | first', result, '执行失败:' + text)

    def tearDown(self):
        self.log.info('-----this is tearDown-----')
        text = '-----step6:删除订阅 expect:成功-----'
        self.log.info(text)
        sql = f"DROP SUBSCRIPTION {self.subname};"
        drop_sub_result = self.commsh_sub.execut_db_sql(
            sql, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE1)
        self.log.info(drop_sub_result)

        text = '-----step7:删除发布 expect:成功-----'
        self.log.info(text)
        sql = f"DROP PUBLICATION {self.pubname};"
        drop_pub_result = self.commsh_pub.execut_db_sql(
            sql, sql_type=self.user_param_pub)
        self.log.info(drop_pub_result)

        text = '-----清理环境,删除表-----'
        self.log.info(text)
        sql = f"drop foreign table {self.tb_name1} cascade;" \
              f"drop table {self.tb_name2} cascade;"
        result = self.commsh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        result = self.commsh_pub.execut_db_sql(sql,
                                               sql_type=self.user_param_pub)
        self.log.info(result)
        guc_res1 = self.commsh_pub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, '',
            f'host    replication  {self.pri_userdb_sub.db_user} '
            f'{self.pri_userdb_sub.db_host}/32')
        self.log.info(guc_res1)
        guc_res = self.commsh_sub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH_REMOTE1,
            f'host    replication  {self.pri_userdb_pub.db_user} '
            f'{self.pri_userdb_pub.db_host}/32',
            macro.DB_ENV_PATH_REMOTE1)
        self.log.info(guc_res)
        pub_mot = self.commsh_pub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f'enable_incremental_checkpoint = on')
        self.log.info(pub_mot)
        sub_mot = self.pri_userdb_sub.sh(
            f'''source {macro.DB_ENV_PATH_REMOTE1};
            gs_guc set -N all -D {macro.DB_INSTANCE_PATH_REMOTE1} 
            -c "enable_incremental_checkpoint=on"''').result()
        self.log.info(sub_mot)
        restart_msg1 = self.commsh_pub.restart_db_cluster()
        self.log.info(restart_msg1)
        restart_msg2 = self.pri_userdb_sub.sh(
            f'''source {macro.DB_ENV_PATH_REMOTE1};
            gs_om -t restart''').result()
        self.log.info(restart_msg2)
        status = self.commsh_pub.get_db_cluster_status()
        self.assertTrue("Degraded" in status or "Normal" in status,
                        '执行失败:' + text)
        self.assertTrue(guc_res, '执行失败:' + text)
        self.assertTrue(guc_res1, '执行失败:' + text)
        self.assertTrue(pub_mot, "执行失败" + text)
        self.assertTrue(sub_mot, "执行失败" + text)
        self.assertIn(self.constant.drop_pub_succ_msg, drop_pub_result,
                      '执行失败' + text)
        self.assertIn(self.constant.drop_sub_succ_msg, drop_sub_result,
                      '执行失败' + text)
        self.log.info(f"-----{os.path.basename(__file__)[:-3]} end-----")
